package cn.azzhu.day05;

import cn.azzhu.day04.ActivityBean;
import cn.azzhu.day04.AsnycGeoToActivityBeanFunction;
import cn.azzhu.utils.FlinkUtilsV1;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisClusterConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.concurrent.TimeUnit;

/**
 * 省份+事件名称+次数[不同维度]的统计结果，写入到MySQL
 * @author azzhu
 * @create 2020-09-19 17:06:26
 */
public class ActivityCount {
    public static void main(String[] args) throws Exception {
        //创建env和添加source
        final DataStream<String> lines = FlinkUtilsV1.createKafkaStream(args, new SimpleStringSchema());

       //关联维度信息，异步IO方式
        //Transformation
        final SingleOutputStreamOperator<ActivityBean> beans = AsyncDataStream.unorderedWait(lines,
                new AsnycGeoToActivityBeanFunction(), 0, TimeUnit.MICROSECONDS, 10);

        //按照不同维度统计
        //Transformation
        final SingleOutputStreamOperator<ActivityBean> summed2 = beans.keyBy("aid","eventType","province").sum("count");

        //调用sink，写入redis
        //创建一个Redis conf
        final FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
        summed2.addSink(new RedisSink<ActivityBean>(conf,new RedisActivityBeanMapper()));

        FlinkUtilsV1.getEnv().execute("ActivityCount");
    }

    public static class RedisActivityBeanMapper implements RedisMapper<ActivityBean> {
        //调用redis的写入方法
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET,"ACT_COUNT");
        }

        //写入Redis中的key
        @Override
        public String getKeyFromData(ActivityBean data) {
            return data.aid+"_"+data.eventType+"_"+data.province;
        }

        @Override
        public String getValueFromData(ActivityBean data) {
            return String.valueOf(data.count);
        }
    }
}
